package com.ferry.registry.service;

import com.ferry.common.util.concurrent.NamedThreadFactory;
import com.ferry.common.util.StackTraceUtil;
import com.ferry.registry.domain.registry.NodeEvent;
import com.ferry.registry.domain.registry.RegistryMetaData;
import com.ferry.registry.domain.registry.RegistryState;
import com.ferry.registry.domain.registry.RegistryValue;
import com.ferry.registry.domain.service.ServiceAddress;
import com.ferry.registry.domain.service.ServiceMetaData;
import com.ferry.registry.listener.NotifyListener;
import com.ferry.registry.listener.OfflineListener;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import io.netty.util.internal.ConcurrentSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.StampedLock;

/**
 * Created by daiyong
 */
public abstract class AbstractRegistryService implements IRegistryService {

	private static final Logger logger = LoggerFactory.getLogger(AbstractRegistryService.class);

	//无界队列 保存注册信息
	private final LinkedBlockingQueue<RegistryMetaData> queue = new LinkedBlockingQueue<>();


	//注册线程
	private final ExecutorService registryExecutor =
			Executors.newSingleThreadExecutor(new NamedThreadFactory("registry.executor", true));

	//注册失败后 定时调度
	private final ScheduledExecutorService registryScheduledExecutor =
			Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("registry.schedule.executor", true));

	//监控线程
	private final ExecutorService localRegisterWatchExecutor =
			Executors.newSingleThreadExecutor(new NamedThreadFactory("local.registry.watch.executor", true));

	private final AtomicBoolean shutdown = new AtomicBoolean(false);

	//注册信息
	private final ConcurrentMap<ServiceMetaData, RegistryValue> registries = Maps.newConcurrentMap();


	//每个服务元数据对应的notify列表
	private final ConcurrentMap<ServiceMetaData, CopyOnWriteArrayList<NotifyListener>> subscribeListener = Maps.newConcurrentMap();

	//每个地址包含的下线监听器列表
	private final ConcurrentMap<ServiceAddress, CopyOnWriteArrayList<OfflineListener>> offlinelistener = Maps.newConcurrentMap();

	// Consumer已订阅的Consumer信息
	private final ConcurrentSet<ServiceMetaData> subscribeSet = new ConcurrentSet<>();

	// Provider已发布的注册信息
	private final ConcurrentMap<RegistryMetaData, RegistryState> registryMetaDataMap = Maps.newConcurrentMap();

	public AbstractRegistryService() {

		//注册线程
		registryExecutor.execute(() -> {
			while (!shutdown.get()) {
				RegistryMetaData meta = null;
				try {
					meta = queue.take();
					registryMetaDataMap.put(meta, RegistryState.DOING);
					doRegistry(meta);
				} catch (InterruptedException e) {
					logger.warn("[registry.executor] interrupted.");
				} catch (Throwable t) {
					if (meta != null) {
						logger.error("Registry [{}] fail: {}, will try again...", meta , StackTraceUtil.stackTrace(t));

						final RegistryMetaData finalMeta = meta;
						registryScheduledExecutor.schedule(() -> {
							queue.add(finalMeta);
						}, 1, TimeUnit.SECONDS);
					}
				}
			}
		});

		//监控线程，每3秒检查一下注册节点
		localRegisterWatchExecutor.execute(() -> {
			while (!shutdown.get()) {
				try {
					Thread.sleep(3000);
					doCheckNodeStatus();
				} catch (InterruptedException e) {
					logger.warn("[local.registry.watch.executor] interrupted.");
				} catch (Throwable t) {
					logger.warn("Check registry node status fail: {}, will try again...", StackTraceUtil.stackTrace(t));
				}
			}
		});
	}

	/**
	 * 取消注册
	 * @param registryMetaData
	 */
	@Override
	public void cancelRegistry(RegistryMetaData registryMetaData) {
		//在队列中 则删除
		if (!queue.remove(registryMetaData)) {
			//否则删除注册节点
			registryMetaDataMap.remove(registryMetaData);
			doCancelRegistry(registryMetaData);
		}
	}

	/**
	 * 订阅服务 添加回调函数 消费者使用
	 * @param serviceMetaData
	 * @param notifyListener
	 */
	@Override
	public void subscribe(ServiceMetaData serviceMetaData, NotifyListener notifyListener) {
		CopyOnWriteArrayList<NotifyListener> notifyListeners = subscribeListener.get(serviceMetaData);
		if (notifyListeners == null) {
			CopyOnWriteArrayList<NotifyListener> notifyListenerNew = new CopyOnWriteArrayList<>();
			notifyListeners = subscribeListener.putIfAbsent(serviceMetaData, notifyListenerNew);
			if (notifyListeners == null) {
				notifyListeners = notifyListenerNew;
			}
		}
		notifyListeners.add(notifyListener);

		subscribeSet.add(serviceMetaData);

		doSubscribe(serviceMetaData);
	}

	/**
	 * 查询注册信息
	 * @param serviceMetaData
	 * @return
	 */
	@Override
	public List<RegistryMetaData> query(ServiceMetaData serviceMetaData) {
		RegistryValue value = registries.get(serviceMetaData);

		if (value == null) {
			return Collections.emptyList();
		}

		final StampedLock stampedLock = value.getLock();

		//时间戳
		final long stamp = stampedLock.readLock();

		try {
			return Lists.newArrayList(value.getMetaSet());
		} finally {
			stampedLock.unlockRead(stamp);
		}
	}

	@Override
	public Map<ServiceMetaData, Integer> consumers() {
		Map<ServiceMetaData, Integer> map = new HashMap<>();

		for (Map.Entry<ServiceMetaData, RegistryValue> entry : registries.entrySet()) {

			RegistryValue value = entry.getValue();
			final StampedLock stampedLock = value.getLock();
			//使用乐观锁 提高效率
			long stamp = stampedLock.tryOptimisticRead();
			int optimisticVal = value.getMetaSet().size();

			//如果value.getMetaSet().size()没有发生变化
			if (stampedLock.validate(stamp)) {
				map.put(entry.getKey(), optimisticVal);
				continue;
			}

			//如果有写操作发生则加悲观锁 重新读取
			stamp = stampedLock.readLock();
			try {
				map.put(entry.getKey(), value.getMetaSet().size());
			} finally {
				stampedLock.unlockRead(stamp);
			}

		}

		return map;
	}

	@Override
	public Map<RegistryMetaData, RegistryState> providers() {
		return new HashMap<>(registryMetaDataMap);
	}

	@Override
	public boolean isShutdown() {
		return shutdown.get();
	}

	@Override
	public void shutdownGracefully() {
		if (!shutdown.getAndSet(true)) {
			try {
				registryExecutor.shutdownNow();
				registryScheduledExecutor.shutdownNow();
				localRegisterWatchExecutor.shutdownNow();
			} catch (Exception e) {
				logger.error("Failed to shutdown: {}.", StackTraceUtil.stackTrace(e));
			} finally {
				doShutDown();
			}
		}
	}

	/**
	 * 添加下线回调接口
	 * @param serviceAddress
	 * @param offlineListener
	 */
	public void addOfflineCallback(ServiceAddress serviceAddress, OfflineListener offlineListener) {
		CopyOnWriteArrayList<OfflineListener> offlineListeners = offlinelistener.get(serviceAddress);
		if (offlineListeners == null) {
			CopyOnWriteArrayList<OfflineListener> newListeners = new CopyOnWriteArrayList<>();
			offlineListeners = offlinelistener.putIfAbsent(serviceAddress, newListeners);
			if (offlineListeners == null) {
				offlineListeners = newListeners;
			}
		}
		offlineListeners.add(offlineListener);
	}

	/**
	 * 下线一次执行下线回调
	 * @param serviceAddress
	 */
	public void offline(ServiceAddress serviceAddress) {
		CopyOnWriteArrayList<OfflineListener> offlineListeners = offlinelistener.remove(serviceAddress);
		if (offlineListeners != null) {
			for (OfflineListener listener : offlineListeners) {
				listener.offline();
			}
		}
	}

	public void offlineListening(ServiceAddress address, OfflineListener listener) {
		CopyOnWriteArrayList<OfflineListener> listeners = offlinelistener.get(address);
		if (listeners == null) {
			CopyOnWriteArrayList<OfflineListener> newListeners = new CopyOnWriteArrayList<>();
			listeners = offlinelistener.putIfAbsent(address, newListeners);
			if (listeners == null) {
				listeners = newListeners;
			}
		}
		listeners.add(listener);
	}

	/**
	 * 通知新增或删除服务
	 * @param serviceMetaData
	 * @param event
	 * @param version
	 * @param array
	 */
	protected void notify(ServiceMetaData serviceMetaData, NodeEvent event, List<RegistryMetaData> registryMetaDatas) {

		RegistryValue value = registries.get(serviceMetaData);
		if (value == null) {
			RegistryValue newValue = new RegistryValue();
			value = registries.putIfAbsent(serviceMetaData, newValue);
			if (value == null) {
				value = newValue;
			}
		}

		final StampedLock stampedLock = value.getLock();
		final long stamp = stampedLock.writeLock();
		try {
			if (event == NodeEvent.LEAF_REMOVED) {
				for (RegistryMetaData registryMetaData : registryMetaDatas) {
					value.getMetaSet().remove(registryMetaData);
				}
			}
			else if (event == NodeEvent.LEAF_ADD) {
				for (RegistryMetaData registryMetaData : registryMetaDatas) {
					value.getMetaSet().add(registryMetaData);
				}
			}
		} finally {
			stampedLock.unlockWrite(stamp);
		}

		CopyOnWriteArrayList<NotifyListener> notifyListeners = subscribeListener.get(serviceMetaData);
		if (notifyListeners != null) {
			for (NotifyListener notifyListener : notifyListeners) {
				for (RegistryMetaData registryMetaData : registryMetaDatas) {
					notifyListener.notify(registryMetaData, event);
				}
			}
		}
	}

	protected ConcurrentSet<ServiceMetaData> getSubscribeSet() {
		return subscribeSet;
	}

	protected ConcurrentMap<RegistryMetaData, RegistryState> getRegistryMetaDataMap() {
		return registryMetaDataMap;
	}

	//注册服务 模板方法 交个具体的来实现
	protected abstract void doRegistry(RegistryMetaData meta);

	//检查注册节点状态
	protected abstract void doCheckNodeStatus();

	//取消注册
	protected abstract void doCancelRegistry(RegistryMetaData meta);

	//关闭注册服务
	public abstract void doShutDown();

	//订阅服务
	public abstract void doSubscribe(ServiceMetaData serviceMetaData);

	@Override
	public void registry(RegistryMetaData registryMetaData) {
		queue.add(registryMetaData);
	}

	public ConcurrentMap<ServiceMetaData, CopyOnWriteArrayList<NotifyListener>> getSubscribeListener() {
		return subscribeListener;
	}

}
